package com.nokia.position;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Hashtable;
import java.util.TimeZone;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.log4j.Logger;

import com.nokia.position.PositionProto.Position;
import com.twitter.elephantbird.mapreduce.io.ProtobufWritable;

public class PositionServiceUsageUSA3Mapper
		extends
		Mapper<LongWritable, ProtobufWritable<Position>, Text, PositionServiceUsageUSA3Writable> {
/*
	private final int MCC_SYSTEM_ID = 0;
	private final int MNC_NETWORK_ID = 1;
	private final int LAC_RNC_TAC_RZ = 2;
	private final int ENB = 3;
	private final int CELLID_BASEID = 4;
	private final int NEIGHBOR_CELL = 5;
	private final int WLAN = 6;
	private final int REFERENCE_LOCATION = 7;
*/
	public static final Logger LOGGER = Logger
			.getLogger(PositionServiceUsageMapper.class.getName());

	Hashtable<String, String> networkTypeMap;
	Hashtable<String, String> platformTypeMap;
	Hashtable<String, String> protocolTypeMap;
	Hashtable<String, String> operatorCountryMap;
	Hashtable<String, String> operatorNameMap;
	Hashtable<String, String> modelTypeMap;

	@Override
	protected void map(LongWritable key, ProtobufWritable<Position> value,
			Context context) throws IOException, InterruptedException {
		// IntWritable One = new IntWritable(1);

		// per date/hour, per country, per network type, per platform, per
		// protocol, per
		// used positioning, per client name
		Position position = value.get();
		/*
		boolean noError = (position.hasStatus() ? position.getStatus() : false);
		
		int eI = 0;
		if (noError) {
			eI = 1;
		}
		long hAI = (position.hasHaccuracy() ? position.getHaccuracy() : 0L);
		*/
		/*
		int usedMeasurements = (position.hasUsedMeasurements() ? position
				.getUsedMeasurements() : 0);

		int b0 = ((usedMeasurements & (1 << MCC_SYSTEM_ID)) > 0 ? 1 : 0);
		int b1 = ((usedMeasurements & (1 << MNC_NETWORK_ID)) > 0 ? 1 : 0);
		int b2 = ((usedMeasurements & (1 << LAC_RNC_TAC_RZ)) > 0 ? 1 : 0);
		int b3 = ((usedMeasurements & (1 << ENB)) > 0 ? 1 : 0);
		int b4 = ((usedMeasurements & (1 << CELLID_BASEID)) > 0 ? 1 : 0);
		int b5 = ((usedMeasurements & (1 << NEIGHBOR_CELL)) > 0 ? 1 : 0);
		int b6 = ((usedMeasurements & (1 << WLAN)) > 0 ? 1 : 0);
		int b7 = ((usedMeasurements & (1 << REFERENCE_LOCATION)) > 0 ? 1 : 0);
		*/
		// WLAN + Neighbor_cell
		//int b8 = ((b6 + b5 == 2) ? 1 : 0);
		// WLAN + CellID BaseID
		//int b9 = ((b6 + b4 == 2) ? 1 : 0);
		
		PositionServiceUsageUSA3Writable psw = new PositionServiceUsageUSA3Writable(1);
		/*
		String usedMeasurementsOutput = b0 + "\t" + b1 + "\t" + b2 + "\t" + b3
				+ "\t" + b4 + "\t" + b5 + "\t" + b6 + "\t" + b7 + "\t" + b8
				+ "\t" + b9;
		*/
		// Operator Name
		String MccMnc = (position.hasServingMccSystemID() ? position
				.getServingMccSystemID() : "")
				+ "\t"
				+ (position.hasServingMncNetworkID() ? position
						.getServingMncNetworkID() : "");
		String operatorName = operatorNameMap.get(MccMnc);
		if (operatorName == null) {
			operatorName = "";
		}

		// Country
		Integer MccInt = (position.hasServingMccSystemID() ? position
				.getServingMccSystemID() : null);

		String country = null;
		if (MccInt != null) {
			country = operatorCountryMap.get(MccInt.toString());
		}
		if (country == null) {
			country = "";
		}

	
		// Platform type
		/*
		Integer platformType = (position.hasClientType() ? position
				.getClientType() : null);
		String platformTypeID = null;
		if (platformType != null) {
			platformTypeID = platformTypeMap.get(platformType.toString());
		}
		*/
		// Protocol type
		Integer protocolType = (position.hasProtocolType() ? position
				.getProtocolType() : null);
		// Model type
		String modelType = (position.hasModelId() ? position
				.getModelId() : null);

		Date dateObj = new Date(position.getCoordinatesTimestamp());
		SimpleDateFormat parser = new SimpleDateFormat("yyyy-MM-dd\tHH");
		parser.setTimeZone(TimeZone.getTimeZone("UTC"));
		String partitionDate = parser.format(dateObj);

		context.write(
				new Text((partitionDate + "\t" 
						+ (MccMnc != null ? MccMnc : "") + "\t" 
						+ (country != null ? country : "") + "\t"  
						+ (protocolType != null ? protocolType : "") + "\t"
						+ (modelType != null ? modelType : "") + "\t"
						+ (operatorName != null ? operatorName : "")
						)),
				psw);
	}

	public void setup(Context context) {

		networkTypeMap = new Hashtable<String, String>();
		platformTypeMap = new Hashtable<String, String>();
		protocolTypeMap = new Hashtable<String, String>();
		operatorCountryMap = new Hashtable<String, String>();
		operatorNameMap = new Hashtable<String, String>();

		try {
			Path[] uris = DistributedCache.getLocalCacheFiles(context
					.getConfiguration());
			for (int i = 0; i < uris.length; i++) {
				if (uris[i].toString().contains("position_network_type.tsv")) {
					try {
						BufferedReader typeMapReader = new BufferedReader(
								new FileReader(uris[i].toString()));
						readTypeMapCache(typeMapReader, this.networkTypeMap);
					} catch (FileNotFoundException e1) {
						e1.printStackTrace();
					}
				} else if (uris[i].toString().contains(
						"position_platform_type.tsv")) {
					try {
						BufferedReader typeMapReader = new BufferedReader(
								new FileReader(uris[i].toString()));
						readTypeMapCache(typeMapReader, this.platformTypeMap);
					} catch (FileNotFoundException e1) {
						e1.printStackTrace();
					}
				} else if (uris[i].toString().contains(
						"position_protocol_type.tsv")) {
					try {
						BufferedReader typeMapReader = new BufferedReader(
								new FileReader(uris[i].toString()));
						readTypeMapCache(typeMapReader, this.protocolTypeMap);
					} catch (FileNotFoundException e1) {
						e1.printStackTrace();
					}
				} else if (uris[i].toString().contains("mcc2country")) {
					try {
						BufferedReader operatorCountryMapReader = new BufferedReader(
								new FileReader(uris[i].toString()));
						readOperatorCountryMapCache(operatorCountryMapReader,
								this.operatorCountryMap);
					} catch (FileNotFoundException e1) {
						e1.printStackTrace();
					}
				} else if (uris[i].toString().contains("MccMncOperatorCountry")) {
					try {
						BufferedReader operatorNameMapReader = new BufferedReader(
								new FileReader(uris[i].toString()));
						readOperatorNameMapCache(operatorNameMapReader,
								this.operatorNameMap);
					} catch (FileNotFoundException e1) {
						e1.printStackTrace();
					}
				} 
				
			}
		} catch (IOException e2) {
			e2.printStackTrace();
		}

	}

	public static void readTypeMapCache(BufferedReader typeMapReader,
			Hashtable<String, String> dim) throws IOException {

		// network_type_id,name,value
		// 1\tWLAN\t1
		// 2\tGSM\t2
		// 3\tWCDMA\t3
		// 4\tLTE\t4
		// 5\tTD-SCDMA\t5
		// 6\tCDMA\t6

		// Key is 3rd column, value (id) is 1st

		// platform_type_id,name,value
		// 1\tService\t1
		// 2\tS40\t2
		// 3\tS60\t3
		// 4\tMeeGo\t4
		// 5\tMt\t5
		// 6\tWP\t6
		// 7\tIE\t7
		// 8\tBing\t8

		// Key is 3rd column, value (id) is 1st

		// protocol_type_id,name,value
		// 1,SUPL-CELLID,1
		// 2,SUPL-AGPS,2
		// 3,SLP-POS,3
		// 4,SLP-GNSS,4
		// 5,SLP-FP,5

		// Key is 3rd column, value (id) is 1st

		String line;
		while ((line = typeMapReader.readLine()) != null) {
			String[] fields = line.toString().split("\t", -1);
			dim.put(fields[2], new String(fields[0]));
		}
	}

	public static void readOperatorCountryMapCache(
			BufferedReader operatorCountryMapReader,
			Hashtable<String, String> dim) throws IOException {

		// 1\t412\tAF

		// Key is 2nd column, value (country) is 3rd

		String line;
		while ((line = operatorCountryMapReader.readLine()) != null) {
			String[] fields = line.toString().split("\t", -1);
			dim.put(fields[1], new String(fields[2]));
		}
	}

	public static void readOperatorNameMapCache(
			BufferedReader operatorNameMapReader, Hashtable<String, String> dim)
			throws IOException {

		// 1\t412\tAF
		// 1504\t310\t970\tGlobalstar USA\tUnited States

		// Key is 2nd and 3rd column, value (operator id) is 1st

		String line;
		while ((line = operatorNameMapReader.readLine()) != null) {
			String[] fields = line.toString().split("\t", -1);
			dim.put(fields[1] + "\t" + fields[2], new String(fields[0]));
		}
	}
	
	public static void readModelTypeMapCache(
			BufferedReader modelTypeMapReader, Hashtable<String, String> dim)
			throws IOException {

		//
		
		String line;
		while ((line = modelTypeMapReader.readLine()) != null) {
			String[] fields = line.toString().split("\t", -1);
			dim.put(fields[1] + "\t" + fields[2], new String(fields[3]));
		}
	}
}
